package middleware.activemq.topic;

import com.me.common.util.PropertiesUtil;
import com.me.common.util.SleepUtil;
import middleware.activemq.queue.QueueProducer;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * 消息的消费者（接受者）
 *
 * @author ME
 * @date 2017/7/24
 */
public class TopicConsumer {
    /**默认连接用户名*/
    private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
    /**默认连接密码*/
    private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
    /**默认连接地址*/
    private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;

    public static void main(String[] args) {
        // 连接工厂
        ConnectionFactory connectionFactory;
        Connection connection = null;
        // 会话 接受或者发送消息的线程
        Session session = null;
        Destination destination;
        // 消息的消费者
        MessageConsumer consumer = null;

        // 实例化连接工厂
        connectionFactory = new ActiveMQConnectionFactory(PropertiesUtil.BROKER_USERNAME, PropertiesUtil.BROKER_PASSWORD, PropertiesUtil.BROKER_URL);

        try {
            // 通过连接工厂获取连接
            connection = connectionFactory.createConnection();
            // 启动连接
            connection.start();
            // 创建session
            session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
            // 创建一个名称为 HelloWorld的消息队列
            destination = session.createTopic("TopicWorld");
            // 创建消息消费者
            consumer = session.createConsumer(destination);

            System.out.println("=====开始监听=====\n");
            while (true) {
                System.out.println("=====主动拉取消息=====");
                // 开一个线程计时
                Thread t1 = new Thread(() -> {
                    for (int i = 1; i < 10; i++) {
                        SleepUtil.seconds(1);
                        System.out.println("计时：" + i);
                    }
                });
                t1.start();

                // 接收数据的时间（等待） 10s  --此处会阻塞进程10s
                TextMessage textMessage = (TextMessage) consumer.receive(10 * 1000);
                if (textMessage != null) {
                    System.out.println("Topic 收到的消息:" + textMessage.getText());
                } else {
                    break;
                }
            }

        } catch (JMSException e) {
            e.printStackTrace();
        } finally {
            QueueProducer.releaseRes(null, consumer, session, connection);
        }

    }
}
